package com.bangcommunity.bbframe.dynconfig.id;

import com.bangcommunity.bbframe.client.zookeeper.ZKClient;
import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import org.apache.curator.RetryPolicy;
import org.apache.curator.RetrySleeper;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Created by tanghc on 17/1/23.
 */
public class ZkId {
    static final Logger logger = LoggerFactory.getLogger(ZkId.class);
    private static ConcurrentMap<String, DistributedAtomicLong> seeds = new ConcurrentHashMap<>();

    private ZkId() {
    }

    private static class ZkIdHolder {
        private static ZkId instance = new ZkId();
    }

    public static ZkId getInstance() {
        return ZkIdHolder.instance;
    }

    /**
     * 获取递增的数值。首次访问时缓存中无数据返回delta。获取失败后等待1ms继续获取,最多等待maxWaitMs。最多尝试次数maxTry
     * 
     * @param key
     *            缓存key值
     * @param delta
     *            递增的步长
     * @param maxWaitMs
     *            最多等待的时长
     * @param maxTry
     *            尝试次数最大值
     */
    public long nextNum(final String key, final long delta, final long maxWaitMs, final int maxTry) {
        logger.debug("start get nextnum: key={}, delta={},maxWaitMs={},maxTry={}", key, delta, maxWaitMs, maxTry);
        Long id = null;
        int trys = 0;
        long expireAt = System.currentTimeMillis() + maxWaitMs;
        while (id == null) {
            if (maxWaitMs > 0 && System.currentTimeMillis() > expireAt) {
                break;
            }
            try {
                DistributedAtomicLong seed = this.getSeed(key);
                if (null == seed) {
                    throw new RuntimeException("no seed for " + key);
                }
                if (null != seed) {
                    AtomicValue<Long> add = seed.add(delta);
                    if (null != add) {
                        id = add.postValue();
                    }
                }
                if (null != id) {
                    return id.longValue();
                }
            } catch (Exception e) {
                logger.error("get nextnum exception: key={}, delta={},maxWaitMs={},maxTry={}", key, delta, maxWaitMs,
                        maxTry, e);
            }
            if (maxTry > 0 && trys++ > maxTry) {
                break;
            }
            try {
                Thread.sleep(1);
            } catch (InterruptedException e) {
            }
        }
        logger.error("get nextnum fail: key={}, delta={},maxWaitMs={},maxTry={}", key, delta, maxWaitMs, maxTry);
        throw new RuntimeException("get next num fail");
    }

    /**
     * 递增id生成 最多等待3秒
     * 
     * @param key
     * @return
     */
    public long nextId(final String key) {
        return nextNum(key, 1, 3 * 1000, -1);
    }

    private DistributedAtomicLong getSeed(String key) throws Exception {

        String seedKeys = getSeedKey(key);
        DistributedAtomicLong distributedAtomicLong = seeds.get(seedKeys);
        if (null == distributedAtomicLong) {
            distributedAtomicLong = new DistributedAtomicLong(ZKClient.getClient(), seedKeys, new RetryPolicy() { // new
                                                                                                                  // RetryNTimes(10,
                                                                                                                  // 10)

                @Override
                public boolean allowRetry(int i, long l, RetrySleeper retrySleeper) {
                    return false;
                }
            });
            // 未重置值
            distributedAtomicLong.initialize(0L);
            seeds.putIfAbsent(seedKeys, distributedAtomicLong);
            distributedAtomicLong = seeds.get(seedKeys);// 保证是统一个seed
        }
        return distributedAtomicLong;
    }

    private String getSeedKey(String key) {

        return new StringBuilder().append("/biz/id/").append(MainConfigLoader.getInstance().getProductCode())
                .append("/").append(MainConfigLoader.getInstance().getAppCode()).append("/").append(key).toString();
    }

}
